package zhao.exmaple.watearkDemo;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

@Slf4j
public class WateSensorUDFmapDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WateSensor> sensorDS = env.fromElements(
                new WateSensor("s1", 1L, 1),
                new WateSensor("s2", 2L, 2),
                new WateSensor("s3", 3L, 3)
        );
//        SingleOutputStreamOperator<WateSensor> so = sensorDS.filter((item) -> {
//            if (item.getId().equals("s1")) {
//                return true;
//            } else {
//                return false;
//            }
//        });
        SingleOutputStreamOperator<WateSensor> so = sensorDS.map(new RichMapFunction<WateSensor, WateSensor>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                String format = StrUtil.format("open 子任务的编号{} 子任务的名称{}", indexOfThisSubtask, getRuntimeContext().getTaskNameWithSubtasks());
                System.out.println(format);

            }

            @Override
            public void close() throws Exception {
                int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                String format = StrUtil.format("close 子任务的编号{} 子任务的名称{}", indexOfThisSubtask, getRuntimeContext().getTaskNameWithSubtasks());
                System.out.println(format);

            }

            @Override
            public WateSensor map(WateSensor value) throws Exception {
                if (value.getId().equals("s1")) {
                    value.setVc(40);
                }
                return value;
            }
        });

//        SingleOutputStreamOperator<WateSensor> so = sensorDS.filter(new WateSensorFilterImpl("s1"));

        so.print();
        env.execute();


    }
}
